package org.apache.ocean.main.merge;

import java.io.File;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.ocean.main.IndexSchema;
import org.apache.ocean.main.http.HttpTimeout;
import org.apache.ocean.main.name.Name;
import org.apache.ocean.main.name.NameClient;
import org.apache.ocean.main.name.Node;
import org.apache.ocean.main.util.CircularQueue;
import org.apache.ocean.main.util.RestartableTimer;
import org.apache.ocean.main.util.Service;
import org.apache.ocean.main.util.Timeout;
import org.apache.ocean.main.util.Util;
import org.apache.ocean.main.util.XMLUtil;
import org.apache.ocean.main.util.RestartableTimer.TimeoutRunnable;
import org.jdom.Element;

public class MergeServer implements Service {
	public static Logger log = Logger.getLogger(MergeServer.class.getName());
	private ExecutorService threadPool;
	private String id;
	private URL url;
	private SearchUrlManager searchUrlManager;
	private CircularQueue<String> cellQueue = new CircularQueue<String>();
	private NameClient nameClient;
	private String serviceStatus;
	private RestartableTimer registerNodeEventTimer;

	public MergeServer() {

	}

	public HttpTimeout getNameHttpTimeout() {
		return new HttpTimeout(1000, 1000);
	}

	public String getServiceStatus() {
		return serviceStatus;
	}

	public URL getUrl() {
		return url;
	}

	public String getId() {
		return id;
	}
  
	// TODO: if contains new cell add to cell queue
	public void nodesUpdated(List<Node> nodes) {
    
	}

	public class RegisterNodeEventTask implements TimeoutRunnable {
		public void run(Timeout timeout) {
			try {
				Name master = nameClient.getMaster(timeout);
				MergeNameClient.registerForNodeEvents(master.getUrl(), "all", 8 * 1000, getNameHttpTimeout());
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public class NameHandler implements NameClient.Listener {
		public void newMaster(Name master) {
			try {
				registerNodeEventTimer.runNow();
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public void start() throws Exception {
		serviceStatus = Service.STARTING;
		File defaultHome = new File(Util.getAppServerHome(), "merge");
		File homeDirectory = Util.getHomeDirectory("oceanmergehome", defaultHome);
		homeDirectory.mkdirs();
		Config config = loadConfig(new File(homeDirectory, "config.xml"));
		List<URL> nameUrls = Util.loadUrls(new File(homeDirectory, "names.txt"));
		nameClient = new NameClient(nameUrls, new NameHandler(), threadPool);
		threadPool = Executors.newFixedThreadPool(config.getThreads(20));
		registerNodeEventTimer = new RestartableTimer("registerNodeEventTimer", new RegisterNodeEventTask(), 10 * 1000, 5 * 1000);
		serviceStatus = Service.STARTED;
	}

	public void stop() throws Exception {
		serviceStatus = Service.STOPPING;
		threadPool.shutdown();
		serviceStatus = Service.STOPPED;
	}

	public void reportError(URL url) {

	}

	public IndexSchema.Loader getIndexSchemaLoader() {
		return nameClient;
	}

	public String getNextWriteableCell() {
		return cellQueue.poll();
	}

	public URL getWriteUrl(String cell) {
		return searchUrlManager.getWriteUrl(cell);
	}

	public List<URL> getSearchUrls() {
    return searchUrlManager.getSearchUrls();
	}

	public static class SearchUrlManager {
		private Map<String,Cell> cellMap = new HashMap<String,Cell>();
		
		public List<URL> getSearchUrls() {
			List<URL> urls = new ArrayList<URL>();
			for (Cell cell : cellMap.values()) {
				URL url = cell.getReadUrl();
				assert url != null;
				urls.add(url);
			}
			return urls;
		}
		
		public URL getWriteUrl(String cellId) {
			Cell cell = cellMap.get(cellId);
			if (cell == null) throw new IllegalArgumentException("cellId: "+cellId+" not found");
			return cell.getWriteUrl();
		}
		
		public void set(String cellId, URL writeUrl, Collection<URL> readUrls) {
			Cell cell = cellMap.get(cellId);
			if (cell == null) {
				cell = new Cell(cellId);
				cellMap.put(cellId, cell);
			}
			cell.set(writeUrl, readUrls);
		}
		
    public static class Cell {
    	private CircularQueue<URL> readUrlQueue = new CircularQueue<URL>();
    	private String cell;
    	private URL writeUrl;
    	
    	public Cell(String cell) {
    		this.cell = cell;
    	}
    	
    	public URL getWriteUrl() {
    		return writeUrl;
    	}
    	
    	public void set(URL writeUrl, Collection<URL> readUrls) {
    		this.writeUrl = writeUrl;
    		readUrlQueue.set(readUrls);
    	}
    	
    	public URL getReadUrl() {
    		return readUrlQueue.poll();
    	}
    }
	}

	public ExecutorService getThreadPool() {
		return threadPool;
	}

	public Config loadConfig(File file) throws Exception {
		Element element = XMLUtil.parseElement(file);
		return new Config(element);
	}

	public static class Config {
		private Integer threads;

		public Config(Element element) {
			threads = XMLUtil.getAttributeInteger("threads", element);
		}

		public int getThreads(int defaultValue) {
			if (threads == null)
				return defaultValue;
			return threads;
		}
	}
}
